package sel.me.popular.library.learning.rocketmq.pull;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

/**
 * 在RocketMQ里，有一种优化的做法——长轮询 Pull
 * ，来平衡推拉模型各自的缺点。基本思路是：消费者如果尝试拉取失败
 * ，不是直接return,而是把连接挂在那里wait,服务端如果有新的消息到来
 * ，把连接notify起来，这也是不错的思路。但海量的长连接block对系统的开销还是不容小觑的
 * ，还是要合理的评估时间间隔，给wait加一个时间上限比较好。
 *
 * https://my.oschina.net/xinxingegeya/blog/956370
 * @author shenenlu 2021年07月07日 下午15:35:36
 */
public class Producer {

    public static void main(String[] args) throws MQClientException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_producer_group_name_5");

        producer.setNamesrvAddr("127.0.0.1:9876");
        producer.start();
        for (int i = 0; i < 1000; i++) {
            try {
                // Create a message instance, specifying topic, tag and message body.
                Message msg = new Message(
                        "PullConsumerTopicTest" /* Topic */,
                        "TagA" /* Tag */,
                        ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
                );
                //Call send message to deliver message to one of brokers.
                SendResult sendResult = producer.send(msg);
                System.out.printf("%s%n", sendResult);
            } catch (Exception e) {
                e.printStackTrace();
                Thread.sleep(1000);
            }
        }
        // Shut down once the producer instance is not longer in use.
        producer.shutdown();
    }

}
